package com.ucode.tool.disruptor;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;

/**
 *  Disruptor生产者
 *  https://github.com/Xlinlin/SpringCloud-Demo/blob/master/SpringCloud-Canal/src/main/java/com/xiao/springcloud/disruptor/DisruptorProducer.java
 *  https://www.cnblogs.com/myJavaEE/p/6790917.html
 *  https://www.cnblogs.com/ohuang/p/5799467.html
 * @author: liliang
 * @date: 2020年2月29日 下午10:20:43
 */
public abstract class DisruptorProducer{

    private static Log log = LogFactory.get();
    
    /**
     * RingBuffer(环形的缓冲区) 大小，必须是 2 的 N 次方(能将求模运算转为位运算提高效率 ), 否则影响性能
     */
    protected int ringBufferSize = 1024 * 1024;
    
    /**
     * disruptor,泛型参数:传递的事件的数据
     */
    protected Disruptor<MessageEvent> disruptor;
    
    /**
     * 是否为多生产者
     */
    protected boolean isMultiProducer;
    
    /**
     *
     * 等待策略->当消费者阻塞在SequenceBarrier上, 消费者如何等待的策略
     * 例如，BlockingWaitStrategy、SleepingWaitStrategy、YieldingWaitStrategy 等，其中，
     * BlockingWaitStrategy 使用锁和条件变量, 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；
     * SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；->多次循环尝试不成功后, 让出CPU, 等待下次调度; 多次调度后仍不成功, 睡眠纳秒级别的时间再尝试. 平衡了延迟和CPU资源占用, 但延迟不均匀.
     * YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。->多次循环尝试不成功后, 让出CPU, 等待下次调度. 平衡了延迟和CPU资源占用, 延迟也比较均匀.
     * BusySpinWaitStrategy 自旋等待，类似自旋锁. 低延迟但同时对CPU资源的占用也多.
     */
    protected WaitStrategy waitStrategy;
    
    /**
     * 环形的缓冲区
     */
    protected RingBuffer<MessageEvent> ringBuffer;
    /**
     * 是否启动
     */
    protected volatile boolean isStart;
    
    /**
     * 关于线程池:
     * https://blog.csdn.net/qq_33300570/article/details/78394188
     */
    protected ExecutorService executor;
    
    protected int threads;
    /**
     * 生产者名称
     */
    protected String producerName; 
    
    public DisruptorProducer(String producerName,boolean isMultiProducer){
        this.threads = Runtime.getRuntime().availableProcessors();//返回可用处理器的Java虚拟机的数量
        //单个生产者
        ProducerType producerType = ProducerType.SINGLE;
        if (isMultiProducer){
            //多生产者
            producerType = ProducerType.MULTI;
        }
        
        // 等待策略
        waitStrategy = new YieldingWaitStrategy();
        
        //初始化disruptor
        /**
         * 第一个参数: 产生Event的工厂类, Event封装生成-消费的数据
         * 第二个参数: RingBuffer的缓冲区大小
         * 第三个参数: 线程池
         * 第四个参数: SINGLE单个生产者, MULTI多个生产者
         * 第五个参数: WaitStrategy 当消费者阻塞在SequenceBarrier上, 消费者如何等待的策略. 
         */
        disruptor = new Disruptor<>(new MessageEventFactory(), ringBufferSize, DisruptorThreadFactory
                .create(producerName+"Disruptor Main-", false), producerType, waitStrategy);
        
        executor = new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), DisruptorThreadFactory
                .create(producerName+"Disruptor Producer-", false), new ThreadPoolExecutor.AbortPolicy());
                
    }
    
    /**
     * 发送事件处理数据
     * @date 2020年3月1日
     * @param dataList 待处理的数据
     */
    public void send(List<Object> dataList){
        if (log.isDebugEnabled()){
            log.debug("批量发送数据给消费者，总数量：{}", CollUtil.isNotEmpty(dataList) ? 0 : dataList.size());
        }
        if (CollUtil.isNotEmpty(dataList)){
            executor.execute(() ->{
                for (Object data : dataList){
                    translator(data);
                }
            });
        }
    }
    
    private void translator(Object data){
        //消息转换类，负责将消息转换为事件
        EventTranslatorOneArg<MessageEvent, Object> eventTranslatorOneArg = (event, sequence, message1) -> event.setData(data);
        //发布
        ringBuffer.publishEvent(eventTranslatorOneArg, data);
    }
    
    /**
     * 停止
     * @date 2020年3月1日
     */
    public void stop(){
        if (isStart){
            //关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
            disruptor.shutdown();
        }
        executor.shutdown();
    }
    
    /**
     * 启动
     * @date 2020年3月1日
     */
    public void start(){
        log.info(">>> "+producerName+ "Disruptor producer 启动...");
        
        setHandleEvents();
        
        setDefaultExceptionHandler();

        ringBuffer = disruptor.start();
        this.isStart = true;
        
        
    }
    
    /**
     * 设置消费者Handler->使用handleEventsWithWorkerPool就可以完成不重复消费，使用handleEventsWith就是重复消费
     * >>>>>disruptor.handleEventsWith()->完成多线程下并发、等待、先后等操作,实现多个消费者并行、顺序重复消费，重复消费类似于kafka中，同一个topic被不同的group的消费者消费
     * 例如：
     * FirstEventHandler firstEventHandler = new FirstEventHandler();
     * SecondEventHandler secondHandler = new SecondEventHandler();
     * ThirdEventHandler thirdEventHandler = new ThirdEventHandler();
     * FourthEventHandler fourthEventHandler = new FourthEventHandler();
     * LastEventHandler lastEventHandler = new LastEventHandler();
     * 一、1,2,last顺序执行
     * disruptor.handleEventsWith(firstEventHandler).handleEventsWith(secondHandler).handleEventsWith(lastEventHandler);
     * 二、1，2，last顺序执行
     * disruptor.handleEventsWith(firstEventHandler);
     * disruptor.after(firstEventHandler).handleEventsWith(secondHandler).then(lastEventHandler);
     * 三、1,2并发执行，之后才是last
     * disruptor.handleEventsWith(firstEventHandler, secondHandler);
     * disruptor.after(firstEventHandler, secondHandler).handleEventsWith(lastEventHandler);
     * 四、1后2，3后4，1和3并发，2和4都结束后last
     * disruptor.handleEventsWith(firstEventHandler, thirdEventHandler);
     * disruptor.after(firstEventHandler).handleEventsWith(secondHandler);
     * disruptor.after(thirdEventHandler).handleEventsWith(fourthEventHandler);
     * disruptor.after(secondHandler, fourthEventHandler).handleEventsWith(lastEventHandler);
     * 
     * >>>>>disruptor.handleEventsWithWorkerPool()->不重复消费，一个消息只能被消费一次
     * 
     * 例如：创建10个消费者(10个线程)来处理同一个生产者发的消息(这10个消费者不重复消费消息)
     * Consumer[] consumers = new Consumer[10];
     * for (int i = 0; i < consumers.length; i++) {
     * consumers[i] = new Consumer();
     * }
     * disruptor.handleEventsWithWorkerPool(consumers);
     * 
     */
    public abstract void setHandleEvents();
    
    /**
     * 设置异常处理
     * @date 2020年3月1日
     */
    public abstract void setDefaultExceptionHandler();

}
